package net.bwie.realtime.jtp.common.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;


public class HbaseUtil {
    private static Connection connection=null;
    static {
        try {
            Configuration conf = HBaseConfiguration.create();
            conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM,"node101:2181,node102:2181,node103:2181");
            conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT,"2181");
            ConnectionFactory.createConnection(conf);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    /**
     * 依据RowKey获取HBase表中数据
     * @param tableName 表名称
     * @param rowKey 主键
     * @param family 列簇名称
     * @param column 列名称
     */
    public static String getData(String tableName, String rowKey, String family, String column)throws Exception{
        Table table = connection.getTable(TableName.valueOf(tableName));
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        byte[] resultValue = result.getValue(Bytes.toBytes(family), Bytes.toBytes(column));
        table.close();
        return Bytes.toString(resultValue);
    }
    /**
     * 扫描HBase表数据
     * @param tableName 表名称
     * @param family 列簇名称
     * @param column 列名称
     */
    public static Map<String, String> scanData(String tableName, String family, String column)throws Exception{

        Table table = connection.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(family));
        ResultScanner scanner = table.getScanner(scan);

        HashMap<String, String> dimMap = new HashMap<>();
        for (Result result : scanner) {
            String rowKey = Bytes.toString(result.getRow());
            String columnValue = Bytes.toString(result.getValue(Bytes.toBytes(family), Bytes.toBytes(column)));
            dimMap.put(rowKey, columnValue);

        }
        scanner.close();
        table.close();
        return dimMap;
    }
    public static void close()throws Exception {
        if (connection != null){
            connection.close();
        }
    }
}
